package com.wjk.kylin.mall.product.kafka.datasync;

import cn.hutool.core.collection.CollectionUtil;
import com.alibaba.fastjson.JSONObject;
import com.wjk.kylin.mall.product.dto.CanalKafkaMessageDTO;
import com.wjk.kylin.mall.product.entity.es.ProductESBean;
import com.wjk.kylin.mall.product.enums.ProductDataSyncEsTableEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;

import java.util.List;
import java.util.Map;

/**
 * @description:
 * @author: wangjinkui
 * @create: 2022-07-18
 **/
@Slf4j
public abstract class ProductDataSyncExecutor implements InitializingBean {
    public abstract ProductDataSyncEsTableEnum getTableEnum();

    /**
     * 新增 事件
     *
     * @param dto
     * @param isExistEsMap
     * @param deleteIdList
     */
    public void insert(CanalKafkaMessageDTO dto, Map<Long, ProductESBean> isExistEsMap, List<Long> deleteIdList) {
        handle(dto, isExistEsMap, deleteIdList, 1);
    }

    /**
     * 编辑 事件
     *
     * @param dto
     * @param isExistEsMap
     * @param deleteIdList
     */
    public void update(CanalKafkaMessageDTO dto, Map<Long, ProductESBean> isExistEsMap, List<Long> deleteIdList) {
        handle(dto, isExistEsMap, deleteIdList, 2);
    }

    /**
     * 删除 事件
     *
     * @param dto
     * @param isExistEsMap
     * @param deleteIdList
     */
    public void delete(CanalKafkaMessageDTO dto, Map<Long, ProductESBean> isExistEsMap, List<Long> deleteIdList) {
        handle(dto, isExistEsMap, deleteIdList, 3);
    }


    /**
     * 新增或更新 es
     *
     * @param fieldsJson
     * @param isExistEsMap
     * @param productId
     */
    public abstract void insertOrUpdateEs(JSONObject fieldsJson, Map<Long, ProductESBean> isExistEsMap, Long productId);

    /**
     * 删除 es
     *
     * @param fieldsJson
     * @param isExistEsMap
     * @param pkNames
     * @param productId
     */
    public abstract void deleteEs(JSONObject fieldsJson, Map<Long, ProductESBean> isExistEsMap, List<String> pkNames, Long productId);

    /**
     * 获取 sku id
     *
     * @param fieldsJson
     * @param deleteIdList
     * @return
     */
    private Long getProductIdForField(JSONObject fieldsJson, List<Long> deleteIdList) {
        Object productIdObj = fieldsJson.get(getTableEnum().getProductIdField());
        if (productIdObj == null) {
            log.error("ProductDataSyncExecutor.getProductIdForField,productId为空,dto:{}", JSONObject.toJSONString(fieldsJson));
            return null;
        }
        Long productId = Long.parseLong(productIdObj.toString());
        if (deleteIdList.contains(productId)) {
            log.info("ProductDataSyncExecutor.getProductIdForField,productId:{} 是需要删除的商品，不需要处理后续逻辑", productId);
            return null;
        }
        return productId;
    }

    /**
     * @param dto
     * @param isExistEsMap
     * @param deleteIdList
     * @param type         1:insert 2:update 3:delete
     */
    private void handle(CanalKafkaMessageDTO dto, Map<Long, ProductESBean> isExistEsMap, List<Long> deleteIdList, int type) {
        List<JSONObject> dataList = dto.getData();
        if (CollectionUtil.isEmpty(dataList)) {
            return;
        }
        for (JSONObject fieldsJson : dataList) {
            Long productId = getProductIdForField(fieldsJson, deleteIdList);
            if (productId == null) {
                continue;
            }
            switch (type) {
                case 1:
                case 2:
                    insertOrUpdateEs(fieldsJson, isExistEsMap, productId);
                    break;
                case 3:
                    deleteEs(fieldsJson, isExistEsMap, dto.getPkNames(), productId);
                    break;
                default:
                    log.error("ProductDataSyncExecutor.handle,type no case,type:{}", type);
                    break;
            }
        }
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        ProductDataSyncExecutorFactory.register(getTableEnum().getValue(), this);
    }
}
